package com.offcn.bigdata.window

import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

/**
 * @Auther: BigData-LGW
 * @ClassName: CountWindow
 * @Date: 2020/12/17 13:25
 * @功能描述: $FunctionDescription
 * @Version:1.0
 */
/**
 * CountWindow
 *  每隔M条记录执行一次window计算
 *  有两种类型：
 *      Tumbling CountWindow 滚动Count
 *          countWindow(M)
 *              不是从源头输入M条记录进行一次计算，而是window中某一个key的纪录条数达到了M，便启动一次window的计算，来计算这些key的次数超过M的数据。
 *      Sliding CountWindow  滑动Count
 *          countWindow(N, M)
 *              窗口中每隔M个元素，就要进行一次计算，窗口中最多存放的元素就是N个
 */
object CountWindowOps {
    def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        val lines = env.socketTextStream("node-1",9999)
        val keyedStream = lines.flatMap(_.split("\\s+"))
            .map((_,1))
            .keyBy(_._1)
        keyedStream.countWindow(4,2)
            .sum(1)
            .print()
        env.execute(s"${CountWindowOps.getClass.getSimpleName}")
    }
}
